/*
 * Name     : FileServer.java
 * Author(s): Jill San Luis
 * Created  : Apr 6, 2011, 12:44:23 AM
 */

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Scanner;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.data.Stat;


/**
 * The FileServer class
 * 
 * @author Jill San Luis
 */
public class FileServer {

	/** Number of partitions */
	public static final int PARTITION_SIZE = 600;

	/** FileServer Primary type */
	public static final int PRIMARY = 0;

	/** FileServer Back Up type */
	public static final int BACKUP = 1;

	/** znode path specific for this class */
	public static final String FILESERVER_PATH = "/fileserver";

	/** Constant representing space character (in String)*/
	public static final String DELIMITER = "#" +
			"";
	
	/** Thread sleep duration */
	private static final int SLEEP_TIME = 5000;
	
	/** File containing the dictionary words */
	private static final String DICTIONARY_FILENAME = "lowercase.rand";

	/** FileServer type */
	private int type = BACKUP;

	/** Partition for Dictionary words */
	private List<String> dictionary;

	/** manager for ZooKeeper server connection */
	private ZkConnector zkc;

	/** Watcher for FileServer zNode */
	private Watcher fileServerWatcher;

	/** Watcher for FilePool zNodes */
	private Watcher filePoolWatcher;
	
	/** Zoo Keeper servers */
	private String zkServers;
//
//	/** Watcher for FilePool request zNodes */
//	private Map<String, Watcher> filePoolWatcherMap;

	private FileServer(){
		
	}
	/**
	 * Creates a new instance of FileServer.
	 * 
	 * @param zkServers	a comma separated list of hosts that can be connected to.
	 */
	private FileServer(String zkServers) {

		this.zkServers  = zkServers;
		initializeDictionary();

		zkc = new ZkConnector();

		try {
			zkc.connect(zkServers);
		} catch (Exception e) {
			System.out.println("Zookeeper connect " + e.getMessage());
		}

		fileServerWatcher = new Watcher() {

			// Anonymous Watcher
			@Override
			public void process(WatchedEvent event) {
				handleEvent(event);
			}

		};

		filePoolWatcher = new Watcher() {
			// Anonymous Watcher
			@Override
			public void process(WatchedEvent event) {
				handleFilePoolUpdate(event);
			}
		};
//
//		filePoolWatcherMap = new HashMap<String, Watcher>();

	}

	/**
	 * Starts the process
	 * 
	 * @param args command line arguments
	 */
	public static void main(String[] args) {

		System.out.println("FileServer starting now");

		if (args.length != 1) {
			System.out
					.println("Usage: java -classpath lib/zookeeper-3.3.2.jar:lib/log4j-1.2.15.jar:. FileServer zkServer:clientPort");

			return;
		}

//		FileServer fs = new FileServer();
//		fs.initializeDictionary();
		FileServer fs = new FileServer(args[0]);
		fs.checkpath();
		

		System.out.println("Spin...");

		while (true) {
			// spin doing nothing!
//
//			try {
//				Thread.sleep(SLEEP_TIME);
//				fs.checkpath();
//			} catch (Exception e) {
//				
//			}
		}
	}

	
	/**
	 * Attaches the personal Worker watcher that will see the changes in this zNode.
	 */
	private void attachWatcher(){
				
		Stat stat = zkc.exists(PoolCreator.FILEPOOL_PATH, filePoolWatcher);

		// zNode doesn't exist; let's try creating it
		if (stat == null) {
			System.out.println("Error: This block should never be reached. Eitherway, process is exiting...");
			System.exit(-1);
		}
		
		System.out.println("Info: Successfully attached watcher.");
	}
	
	private void attachFilePoolWatcher(String zkServers) {

		PoolCreator.createPool(zkc, PoolCreator.FILEPOOL_PATH);

		Stat stat = zkc.exists(PoolCreator.FILEPOOL_PATH, false); 
		if (stat == null) {
			System.out.println("Failed to create " + PoolCreator.FILEPOOL_PATH + ". Exiting.");
			System.exit(-1);
		}
		else{
			
			try{
				List<String> childrenZNode = zkc.getZooKeeper().getChildren(PoolCreator.FILEPOOL_PATH, filePoolWatcher);
				handleChildren(childrenZNode);
			}
			catch(KeeperException ke){
				System.out.println("Failed to attach filePoolWatcher "  + ke.getMessage());
			}
			catch(InterruptedException ie){
				System.out.println("Failed to attach filePoolWatcher "  + ie.getMessage());
			}
			
			System.out.println("Info: Successfully attached File Pool Watcher.");
		}		
	}

	/**
	 * Checks if the path to this zNode exist already. This File Server acquires
	 * the zNode if it is not owned by other File Server already.
	 */
	private void checkpath() {
		Stat stat = zkc.exists(FILESERVER_PATH, fileServerWatcher);

		// zNode doesn't exist; let's try creating it
		if (stat == null) {
			System.out.println("Creating " + FILESERVER_PATH);

			Code ret = zkc.create(FILESERVER_PATH, // Path of zNode
					null, // Data
					CreateMode.EPHEMERAL // zNode type, set to EPHEMERAL.
					);

			if (ret == Code.OK) {
				System.out.println("PRIMARY File Server process");
				type = PRIMARY;
			} 
		}
		
		if(type != PRIMARY) {
			System.out.println("BACKUP File Server process");
			type = BACKUP;
		}
		
		attachFilePoolWatcher(zkServers);
	}

	/**
	 * Handles partition retrieval requests from Workers.
	 * 
	 * @param event	ZooKeeper watched event
	 */
	private void handleFilePoolUpdate(WatchedEvent event) {

		System.out.println("Info: handleFilePoolUpdate is starting.");
		if (PoolCreator.FILEPOOL_PATH.equals(event.getPath())
				&& event.getType() == Watcher.Event.EventType.NodeChildrenChanged) {

			System.out.println("Info: Updating watchers for each FileServer request");
			try{
				// update the map of children watchers
				List<String> childrenZNode = zkc.getZooKeeper().getChildren(
						PoolCreator.FILEPOOL_PATH, filePoolWatcher);
					
				handleChildren(childrenZNode);
			}
			catch(KeeperException ke){
				// do nothing
				System.out.println("Error: Cannot retrieve childrenZNodes - " + ke.getMessage());
			}
			catch(InterruptedException ie){
				// do nothing
				System.out.println("Error: Cannot retrieve childrenZNodes - " + ie.getMessage());
			}
		}
		
		System.out.println("Info: handleFilePoolUpdate is done.");
		
		attachWatcher();
	}

	private void handleChildren(List<String> childrenZNode) {
		System.out.println("Received children " + childrenZNode.size());
		List<String> deletableNodes = new ArrayList<String>();
		
		for (String czn : childrenZNode) {
			
			// just read each one of them and cater to it
			if(type == PRIMARY){
				
				String fullPath = PoolCreator.FILEPOOL_PATH + "/" + czn;
				handlePartitionRequest(fullPath);
				deletableNodes.add(fullPath);
				System.out.print("Info: Added " + fullPath + " to deletable paths.");
			}
		}

		deleteNodes(deletableNodes);
		
	}
	private void deleteNodes(List<String> deletableNodes) {
		
		System.out.println("Deleting already processed requests.");
		
		for(String nodePath: deletableNodes){
		
			try{
				Stat s = zkc.exists(nodePath, false);
			
				if(s!=null){
					
					System.out.println("Deleting " + nodePath);
					zkc.getZooKeeper().delete(nodePath, -1);
				}
			}
			catch(KeeperException ke){
				System.out.println("Warning: Exception when deleting " + nodePath + " " + ke.getMessage());
			}
			catch(InterruptedException ie){
				System.out.println("Warning: Exception when deleting " + nodePath + " " + ie.getMessage());
			}
		}
		
		System.out.println("Finished deletion process.");
		
	}

	/**
	 * @param event	ZooKeeper watched event
	 */
	private void handlePartitionRequest(String path) {
		
		System.out.println("Info: PRIMARY is processing partition request from " + path);
		
		try {
			Stat readStat = new Stat();
			byte[] partitionRequestByte = zkc.getZooKeeper().getData( path,
					false, readStat);
			
			System.out.println(partitionRequestByte);
	
			String partitionRequest = new String(partitionRequestByte);
			
			System.out.println(partitionRequest);
									
			// token[0] - worker path
			// token[1] - partition ID
			String[] token = partitionRequest.split(ZkConnector.DATADELIMITER);
			
			if(token.length != 2){
				System.out.println("Error: Invalid partition request format.");
				return;
			}
			
			try{
				int partitionIndex = Integer.parseInt(token[1]);
				String data = partitionRequest + ZkConnector.DATADELIMITER + dictionary.get(partitionIndex);
				zkc.atomicAppend(data, token[0]);				
				System.out.println("Info: Sending [" + data + "] to [" + token[0] + "]");
			}
			catch(NumberFormatException nfe){
				System.out.println("Error: Cannot proceed with partition fetching - invalid partition ID " + token[0]);
			}
			

		} catch (KeeperException ke) {

			// do something about it
			System.out.println("Node Data Changed " + ke.getMessage());
		} catch (InterruptedException ie) {

			// do something about it
			System.out.println("Node Data Changed " + ie.getMessage());
		}
		
		System.out.println("Finished handling partition request.");



	}

	

	/**
	 * @param event	ZooKeeper watched event
	 */
	private void handlePartitionRequest(WatchedEvent event) {
		
		System.out.println("Info: handlePartitionRequest");

		if (type == PRIMARY	 && event.getType() == EventType.NodeDataChanged ) {
			
			try {
				
				Stat getDataStat = new Stat();
				byte[] partitionRequestByte = zkc.getZooKeeper().getData(event.getPath(),
						false, getDataStat);
				
				if(partitionRequestByte != null){
					String partitionRequest = new String(partitionRequestByte);
					System.out.println(partitionRequest);
											
					// token[0] - worker path
					// token[1] - partition ID
					String[] token = partitionRequest.split(ZkConnector.DATADELIMITER);
					
					if(token.length != 2){
						System.out.println("Error: Invalid partition request format.");
						return;
					}
					
					try{
						int partitionIndex = Integer.parseInt(token[1]);
						String data = partitionRequest + ZkConnector.DATADELIMITER + dictionary.get(partitionIndex);
						zkc.atomicAppend(data, token[0]);
						
						System.out.println("Info: Sending [" + data + "] to [" + token[0] + "]");
					}
					catch(NumberFormatException nfe){
						System.out.println("Error: Cannot proceed with partition fetching - invalid partition ID " + token[0]);
					}
				
				}
				else{
					System.out.println("Data in the child FilePool zNode is NULL.");
				}
				

			} catch (KeeperException ke) {

				// do something about it
				System.out.println("Node Data Changed " + ke.getMessage());
			} catch (InterruptedException ie) {

				// do something about it
				System.out.println("Node Data Changed " + ie.getMessage());
			}

		}

	}

	/**
	 * Handles events related to File Server process - zNode deleted, zNode
	 * created.
	 * 
	 * @param event
	 *            ZooKeeper watched event
	 */
	private void handleEvent(WatchedEvent event) {
		String path = event.getPath();
		Watcher.Event.EventType eventType = event.getType();
		
		System.out.println("Info: Received event " + eventType + " with path " + path + " comparing against " + FILESERVER_PATH);

		if (FILESERVER_PATH.equalsIgnoreCase(path)) {

			// Existing FileServer zNode has been deleted. Try to take over!
			if (eventType == Watcher.Event.EventType.NodeDeleted) {
				System.out.println(FILESERVER_PATH + " deleted! Chase after the node.");
				checkpath();
			}

			// FileServer zNode has been created. Re-enable the watch for this
			// node.
			else if (eventType == Watcher.Event.EventType.NodeCreated) {
				System.out.println(FILESERVER_PATH + " created!");
				checkpath();
			}
		}
	}

	/**
	 * Partitions the Dictionary into sections.
	 */
	private void initializeDictionary() {
		
		
		dictionary = new ArrayList<String>();
		for(int i=0; i< PARTITION_SIZE; i++){
			dictionary.add(DELIMITER);
		}
		
		
		// distribute the words in each partition as equally as possible
		Scanner scanner = new Scanner(FileServer.class
				.getResourceAsStream(DICTIONARY_FILENAME));

		int counter = 0;

		int firstLayer = 0;
		while (scanner.hasNext()) {

			int index = counter%PARTITION_SIZE;
			counter++;
			
			if(firstLayer < PARTITION_SIZE){
				dictionary.set(index, scanner.next());
				firstLayer++;
				continue;
			}
			
			String section = dictionary.get(index) + DELIMITER +  scanner.next();
			dictionary.set(index, section);			
		}		
	}
}
